package com.codejiwei.core.graphx

import org.apache.spark.SparkContext
import org.apache.spark.graphx._

object GraphX_AggregateMessages {
  def main(args: Array[String]): Unit = {
    // 创建 SparkContext
    val sc = new SparkContext("local", "GraphAggregateMessagesExample")

    // 创建一个简单的图
    val vertices = sc.parallelize(Array((1L, "Alice"), (2L, "Bob"), (3L, "Charlie"), (4L, "David")))
    val edges = sc.parallelize(Array(
      Edge(1L, 2L, 3), // 边1: Alice -> Bob，权重为3
      Edge(1L, 2L, 2), // 边2: Alice -> Bob，权重为2
      Edge(1L, 3L, 1), // 边3: Alice -> Charlie，权重为1
      Edge(2L, 3L, 4), // 边4: Bob -> Charlie，权重为4
      Edge(3L, 4L, 5), // 边5: Charlie -> David，权重为5
      Edge(3L, 4L, 6) // 边6: Charlie -> David，权重为6
    ))

    val graph: Graph[String, Int] = Graph(vertices, edges, "default")

    // 在每条边上发送消息，并将消息合并成一个单一的消息
    val aggregatedMessages = graph.aggregateMessages[Int](
      sendMsg = ctx => ctx.sendToDst(ctx.attr), // 在每条边上发送边的属性值给目标顶点
      mergeMsg = (a, b) => math.max(a, b), // 将多条消息合并为单一消息，这里取最大值
      tripletFields = TripletFields.All // 考虑所有三元组字段
    )

    // 输出每个顶点的聚合结果
    aggregatedMessages.collect().foreach { case (vertexId, aggregatedValue) =>
      println(s"Vertex $vertexId's aggregated value: $aggregatedValue")
    }
  }
}
